import os
import re
import subprocess
import time
import xml.dom.minidom
from typing import Set, Union, List, Dict, Iterable, Callable

from scripts.Log import Log
from scripts.config.Config import Config
from scripts.device.Device import Device


class FileInfo(object):  # from ls -all
    PATTERN_LINE = re.compile(
        r'(\w)([rwx-]{9})\s+\d+\s+(\w+)\s+(\w+)+\s+(\d+)\s+(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}\.\d+ [-+]\d{4})\s+(.+)')

    # 1: type
    # 2: permission
    # 3: user
    # 4: group
    # 5: size
    # 6: datetime
    # 7: package name

    def __init__(self, info_str: str):
        self.raw = info_str
        matched = FileInfo.PATTERN_LINE.match(info_str)
        if matched:
            self.type = matched.group(1)
            self.permission = matched.group(2)
            self.user = matched.group(3)
            self.group = matched.group(4)
            self.size = matched.group(5)
            self.datetime = matched.group(6)
            self.pkg_name = matched.group(7)
        else:
            self.type = None
            self.permission = None
            self.user = None
            self.group = None
            self.size = None
            self.datetime = None
            self.pkg_name = None

    def is_valid(self):
        return self.pkg_name is not None


class PkgInfo(object):
    PATTERN_USERID = re.compile(r'userId=(\d+)\s*')
    PATTERN_PKG = re.compile(r'pkg=Package{.+?\s+(.+)}\s*')
    PATTERN_LOCATION = re.compile(r'resourcePath=(.+)\s*')
    PATTERN_DATA_DIR = re.compile(r'dataDir=(.+)\s*')
    PATTERN_FLAGS = re.compile(r'pkgFlags=\[(.+)]\s*')
    PATTERN_VERSION = re.compile(r'versionCode=(\d+)\s+minSdk=(\d+)\s+targetSdk=(\d+)\s*')

    def __init__(self, raw: str):
        self.raw = raw

        lines = self.raw.split('\n')
        for line in lines:
            matched = PkgInfo.PATTERN_USERID.match(line)
            if matched:
                self.uid = matched.group(1)
                continue

            matched = PkgInfo.PATTERN_PKG.match(line)
            if matched:
                self.pkg = matched.group(1)
                continue

            matched = PkgInfo.PATTERN_LOCATION.match(line)
            if matched:
                self.location = matched.group(1)
                continue

            matched = PkgInfo.PATTERN_DATA_DIR.match(line)
            if matched:
                self.data_dir = matched.group(1)
                continue

            matched = PkgInfo.PATTERN_FLAGS.match(line)
            if matched:
                flags = [f.strip() for f in matched.group(1).split()]
                self.flags = list(filter(lambda f: f != '', flags))
                self.allow_backup = 'ALLOW_BACKUP' in self.flags
                self.restore_any_version = 'RESTORE_ANY_VERSION' in self.flags
                continue

            matched = PkgInfo.PATTERN_VERSION.match(line)
            if matched:
                self.version_code = matched.group(1)
                self.min_sdk = matched.group(2)
                self.target_sdk = matched.group(3)


class WindowPolicy(object):
    def __init__(self, raw: List[str]):
        self.raw = raw

        pattern_keyguard_label = re.compile(r'\s*KeyguardServiceDelegate\s*')
        pattern_keyguard_showing = re.compile(r'\s*showing=(\w+)\s*')
        pattern_screen_state = re.compile(r'\s*screenState=(\w+)\s*')

        found_keyguard_label = False
        for line in self.raw:
            if found_keyguard_label:
                if pattern_keyguard_label.match(line):
                    found_keyguard_label = True
            else:
                matched = pattern_keyguard_showing.match(line)
                if matched:
                    self.keyguard_showing = matched.group(1) == 'true'
                    continue

                matched = pattern_screen_state.match(line)
                if matched:
                    self.screen_on = matched.group(1) == 'SCREEN_STATE_ON'
                    continue


class PmListUserInfo(object):
    def __init__(self, user_id: int, user_name: str, flags: str):
        self.user_id = user_id
        self.user_name = user_name
        self.flags = flags


class PmListPkgInfo(object):
    def __init__(self, pkg: str, uid: int, apk_path: str):
        self.pkg = pkg
        self.uid = uid
        self.apk_path = apk_path


class AdbShell(object):
    ADB_BIN = os.path.join('bin', 'adb.exe')
    _BIN: Union[None, str] = None

    DUMP_EXEC = Config.ADB_DEVICE_DUMP_EXEC

    ERROR_CODE_OFFLINE = -10000
    ERROR_CODE_OFFLINE_ON_FINISH = -10086

    RESULT_TAG = '&=#R=#&:'
    PATTERN_RESULT = re.compile(r'(.*)%s(\d+)\s*' % RESULT_TAG)

    PATTERN_PM_LIST_PACKAGES = re.compile(r'\s*package:\s*(.+)\s*=\s*(.+)\s+uid:\s*(\d+)\s*')

    PATTERN_FIND_ERROR = re.compile(r'find:\s+.+:\s+No such file or directory')
    PATTERN_PM_LIST_USER = re.compile(r'\s*UserInfo{(\d+):(.+):(.+)}.*')

    PATTERN_ADB_DAEMON_START = re.compile(r'\s*\* daemon not running;.*')
    PATTERN_ADB_DAEMON_STARTED = re.compile(r'\s*\* daemon started successfully.*')

    STDIN_BUFF_SIZE = 1024 * 1024  # 1MB

    PATTERN_TAR_EMPTY_ARCHIVE = re.compile(r'tar:\s+empty\s+archive\s*')
    TAR_ERROR_REASON_UNKNOWN = 0
    TAR_ERROR_REASON_EMPTY = 1

    SERVICE_SETTINGS = 'settings'
    SERVICE_PACKAGE = 'package'
    SERVICE_ACTIVITY = 'activity'
    SERVICE_INPUT = 'input'
    SERVICE_LOCK_SETTINGS = 'lock_settings'
    SERVICE_BLUETOOTH = 'bluetooth_manager'
    SERVICE_PHONE = 'phone'
    SERVICE_NFC = 'nfc'
    SERVICE_POWER = 'power'
    SERVICE_WINDOW = 'window'
    SERVICE_UIMODE = 'uimode'
    SERVICE_SURFACEFLINGER = 'SurfaceFlinger'

    PATTERN_WM_DENSITY = (
        re.compile(r'(Physical)\s+density:\s+(\d+)\s*'),
        re.compile(r'(Override)\s+density:\s+(\d+)\s*')
    )

    NIGHT_MODE_OFF = 1
    NIGHT_MODE_ON = 2
    NIGHT_MODE_AUTO = 3

    def __init__(self, sn: str):
        self.sn = sn
        self.stdin: Union[None, Device.ShellIn] = None
        self.stdout: Union[None, Device.ShellOut] = None
        self.proc: Union[None, subprocess.Popen] = None
        self.enter_count = 0

        self.ready_services: Set[str] = set()

    def __enter__(self):
        if self.enter_count == 0:
            self.start()
        self.enter_count += 1
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.enter_count -= 1
        if self.enter_count <= 0:
            self.exit()

    @staticmethod
    def bin() -> str:
        if AdbShell._BIN is None:
            AdbShell._BIN = os.path.join(Device.script_dir(), AdbShell.ADB_BIN)
        return AdbShell._BIN

    def _require_service(self, service: Union[str, Iterable[str]], retry: int = -1) -> bool:
        return self.wait_for_service(service, retry)

    def wait_for_service(self, service: Union[str, Iterable[str]], retry: int = -1) -> bool:
        if isinstance(service, str):
            if service in self.ready_services:
                return True

            while True:
                ready = self.service_check(service)
                if ready or retry == 0:
                    break

                Log.d('service [%s] not ready, recheck after 1s!!' % service)
                retry -= 1
                time.sleep(1)

            if ready:
                self.ready_services.add(service)
            return ready
        elif isinstance(service, Iterable):
            result = True
            for s in service:
                result &= self._require_service(s, retry)
            return result

    def _reset_service_status(self, service: Union[None, str, List[str]] = None):
        if service is None:
            self.ready_services.clear()
        elif isinstance(service, str):
            if service in self.ready_services:
                self.ready_services.remove(service)
        elif isinstance(service, Iterable):
            for s in service:
                self._reset_service_status(s)

    def start(self):
        self.stdin, self.stdout, self.proc = Device.async_shell('"%s" -s %s shell' % (AdbShell.bin(), self.sn))

    def restart(self):
        self.ready_services.clear()
        self.start()

    def exit(self):
        self.stdin.writeline('exit')
        try:
            self.stdin.flush()
        except OSError:
            pass
        self.proc.wait()

    def wait_for_shell_exit(self, timeout: int):
        try:
            self.proc.wait(timeout=timeout)
            return True
        except subprocess.TimeoutExpired:
            return False

    @staticmethod
    def _on_input_callback(on_input: Union[Callable[[Device.ShellIn], None]], stdin: Device.ShellIn):
        on_input(stdin)
        stdin.flush()

    def _exec(self, cmd: str, on_input: Union[None, Callable[[Device.ShellIn], None]] = None) -> (int, List[str]):
        if self.proc.poll() is not None:
            Log.w('shell already disconnected')
            return AdbShell.ERROR_CODE_OFFLINE, self.stdout.readlines()

        # send cmd
        if AdbShell.DUMP_EXEC:
            Log.d('exec: << (%s) 2>&1' % cmd)  # dump exec
        self.stdin.writeline('(%s) 2>&1' % cmd)
        self.stdin.flush()

        # send input
        if on_input:
            from scripts.util.Util import Util
            Util.ignore_exception(lambda: AdbShell._on_input_callback(on_input, self.stdin))

        # get cmd result
        if AdbShell.DUMP_EXEC:
            Log.d('exec: << echo "%s$?"' % AdbShell.RESULT_TAG)  # dump exec
        self.stdin.writeline('echo "%s$?"' % AdbShell.RESULT_TAG)
        self.stdin.flush()

        # parse output
        lines: List[str] = []
        code: int
        while True:
            line = self.stdout.readline()
            if AdbShell.DUMP_EXEC:
                Log.d('exec: >> %s' % line)  # dump exec

            # check for disconnected
            if line == '':
                # device disconnected
                Log.w('shell disconnected')
                code = AdbShell.ERROR_CODE_OFFLINE_ON_FINISH
                break

            # collect output or get the result
            matched = AdbShell.PATTERN_RESULT.fullmatch(line)
            if matched:
                if matched.group(1) != '':
                    lines.append(matched.group(1))
                code = int(matched.group(2))
                break
            else:
                lines.append(line)

        return code, AdbShell.ignore_adb_daemon_start_info(lines)

    def exec(self, cmd: str, on_input: Union[None, Callable[[Device.ShellIn], None]] = None,
             retry: int = 3, allow_offline_on_finish: bool = False) -> (int, List[str]):
        code, lines = self._exec(cmd, on_input)
        while code != 0 and retry != 0:
            Log.w('AdbShell.exec()# cmd [%s] failed. retry after 1 second' % cmd)
            Log.w('AdbShell.exec()# code: %d, lines: %s' % (code, lines))
            retry -= 1
            time.sleep(1)

            # check for offline
            if code == AdbShell.ERROR_CODE_OFFLINE:
                Log.w('reconnect for offline')
                self._reconnect_device()
            if code == AdbShell.ERROR_CODE_OFFLINE_ON_FINISH:
                if allow_offline_on_finish:
                    Log.w('ignore offline on finish due to allow_offline_on_finish is True')
                    return 0, lines
                else:
                    Log.w('reconnect for offline on finish')
                    self._reconnect_device()

            code, lines = self._exec(cmd, on_input)
            if code == 0:
                Log.d('AdbShell.exec()# cmd [%s] success after retry' % cmd)

        # check code
        if code != 0:
            Log.e('AdbShell.exec()# failed to execute cmd [%s]' % cmd)
            Log.e('AdbShell.exec()# code: %d, lines: %s' % (code, lines))

        return code, lines

    def _reconnect_device(self):
        from scripts.device.AdbDevice import AdbDevice
        device = AdbDevice.wait_for_device(self.sn)  # wait for device online
        device.adb_root()  # root again
        AdbDevice.wait_for_device(self.sn)  # wait for device online again
        self.restart()  # restart shell
        self.wait_for_filesystem_ready()  # wait for filesystem ready

    def exec_on_require(self, service: Union[str, Iterable[str]], cmd: str, retry_code: int = 20,
                        require_boot_complete: bool = False, require_filesystem_ready: bool = False,
                        on_input: Union[None, Callable[[Device.ShellIn], None]] = None,
                        retry: int = 3, allow_offline_on_finish: bool = False) -> (int, List[str]):
        if require_filesystem_ready:
            self.wait_for_filesystem_ready()
        if require_boot_complete:
            self.wait_for_boot_complete()
        self._require_service(service)
        code, lines = self.exec(cmd, on_input, retry, allow_offline_on_finish)
        if code == retry_code:  # cmd: Can't find service: XXXXX
            if require_filesystem_ready:
                self.wait_for_filesystem_ready()
            if require_boot_complete:
                self.wait_for_boot_complete()
            self._reset_service_status(service)
            self._require_service(service)
            code, lines = self.exec(cmd, on_input, retry, allow_offline_on_finish)
            if code == 0:
                Log.d('AdbShell.exec_on_require()# cmd [%s] success after retry' % cmd)
        return code, lines

    @staticmethod
    def ignore_adb_daemon_start_info(lines: List[str]) -> List[str]:
        if len(lines) >= 2 and \
                AdbShell.PATTERN_ADB_DAEMON_START.match(lines[0]) and \
                AdbShell.PATTERN_ADB_DAEMON_STARTED.match(lines[1]):
            return lines[2:]
        else:
            return lines

    def reboot_bootloader(self):
        self.exec('reboot bootloader')

    def reboot(self):
        self.exec('reboot', retry=0, allow_offline_on_finish=True)

    def wait_for_filesystem_ready(self, user: int = 0):
        while not self.path_exists('/data/user/%d/android' % user):
            Log.d('file system not ready, retry after 1 second')
            time.sleep(1)

    def wait_for_path_ready(self, path: str):
        while not self.path_exists(path):
            Log.d('file [%s] not ready, retry after 1 second' % path)
            time.sleep(1)

    def shell_uid(self) -> Union[None, int]:
        code, lines = self.exec('id -u')
        if code == 0:
            return int(lines[0])
        else:
            return None

    def is_rooted(self):
        return self.shell_uid() == 0

    def android_start(self):
        self.exec('start')

    def android_stop(self):
        self.exec('stop')
        self._reset_service_status()

    def path_exists(self, path: str) -> bool:
        _, lines = self.exec('if [ -e "%s" ]; then echo true; else echo false; fi' % path)
        if len(lines) == 0:
            return False
        return lines[0].strip() == 'true'

    def path_is_dir(self, path: str) -> bool:
        _, lines = self.exec('if [ -d "%s" ]; then echo true; else echo false; fi' % path)
        if len(lines) == 0:
            return False
        return lines[0].strip() == 'true'

    def ls(self, path: str) -> List[str]:
        _, lines = self.exec('ls %s' % path)
        return [line.strip() for line in lines]

    def ls_all(self, path: str) -> List[FileInfo]:
        _, lines = self.exec('ls -all %s' % path)
        infos = [FileInfo(line.strip()) for line in lines]
        return list(filter(lambda info: info.is_valid() and info.pkg_name not in ('.', '..'), infos))

    def getprop(self, key: str) -> Union[None, str]:
        _, lines = self.exec('getprop %s' % key)
        if len(lines) > 0:
            return lines[0].strip()
        else:
            return None

    def setprop(self, key: str, value: str):
        self.exec('setprop %s %s' % (key, value))

    def android_version(self) -> int:
        version = self.getprop('ro.build.version.release')
        return int(version) if version is not None else 0

    def device_model(self) -> str:
        model = self.getprop('ro.product.model')
        return model if model is not None else ''

    def mkdir(self, path: str) -> bool:
        code, _ = self.exec('mkdir -p "%s"' % path)
        return code == 0

    def is_oem_unlock_allowed(self) -> bool:
        _, lines = self.exec('getprop sys.oem_unlock_allowed')
        if len(lines) <= 0:
            return False
        return lines[0] == '1'

    def mkdir_and_check(self, path: str):
        while not self.mkdir(path):
            Log.d('create dir failed. retry after 1 second!')
            time.sleep(1)

    def rm(self, path: str, recursive=True):
        if recursive:
            self.exec('rm -rf "%s"' % path)
        else:
            self.exec('rm -f "%s"' % path)

    def tar_compress(self, src: str, dst: str,
                     include: Union[None, List[str]], exclude: Union[None, List[str]] = None):
        src_dir = os.path.dirname(src)
        src_name = os.path.basename(src)

        include_cmd = ' '.join(['\'%s/%s\'' % (src_name, i) for i in include]) if include is not None else src_name
        exclude_cmd = ' '.join(
            ['--exclude=\'%s/%s\'' % (src_name, i) for i in exclude]) if exclude is not None else ''

        retry = 3
        while True:
            retry -= 1
            code, lines = self.exec('cd "%s"; tar -zcf "%s" %s %s' % (src_dir, dst, include_cmd, exclude_cmd),
                                    retry=0)

            result = code == 0
            reason = AdbShell.TAR_ERROR_REASON_UNKNOWN
            if code != 0 and len(lines) > 0:
                if AdbShell.PATTERN_TAR_EMPTY_ARCHIVE.fullmatch(lines[0]) is not None:
                    reason = AdbShell.TAR_ERROR_REASON_EMPTY

            if result or reason == AdbShell.TAR_ERROR_REASON_EMPTY or retry == 0:
                return result, reason

            Log.w('tar_compress failed, retry...')

    def tar_extract(self, src: str, dst: str):
        self.exec('cd "%s"; tar -zxf "%s"' % (dst, src))

    def pkg_info(self):
        _, lines = self.exec_on_require(AdbShell.SERVICE_PACKAGE,
                                        'pm list packages -f')
        line_pattern = re.compile(r'package:(.+)=(.+)')
        return {matched.group(2): matched.group(1) for matched in [line_pattern.match(line) for line in lines]}

    def dump_pkg(self) -> Dict[str, PkgInfo]:
        _, lines = self.exec('dumpsys package')
        raw = '\n'.join([line.strip() for line in lines])

        pattern = re.compile(r'[\s\S]*\s+Packages:\s+([\s\S]+)\n\n[\s\S]*')
        matched = pattern.match(raw)
        if not matched:
            return dict()
        raw_pkg_infos = matched.group(1)

        infos = dict()
        pattern = re.compile(r'Package \[[^\n]+\n')
        matched = pattern.finditer(raw_pkg_infos)
        begin = -1
        for i in matched:
            if begin < 0:
                begin = i.regs[0][1]
                continue
            info = PkgInfo(raw_pkg_infos[begin: i.regs[0][0]])
            if hasattr(info, 'pkg'):
                infos[info.pkg] = info
            else:
                Log.w('Ignore invalid pkg')
            begin = i.regs[0][1]
        info = PkgInfo(raw_pkg_infos[begin: len(raw_pkg_infos)])
        infos[info.pkg] = info
        return infos

    def chown(self, user: Union[int, str], group: Union[int, str], path: str, recursive=True):
        if recursive:
            self.exec('chown -R %s:%s "%s"' % (user, group, path))
        else:
            self.exec('chown %s:%s "%s"' % (user, group, path))

    def chown_on_exist(self, user: Union[int, str], group: Union[int, str], path: str, recursive=True):
        cmds = [
            'if [ -e "%s" ]; then' % path,
            'chown -R %s:%s "%s"' % (user, group, path) if recursive else 'chown %s:%s "%s"' % (user, group, path),
            'fi'
        ]
        self.exec('\n'.join(cmds))

    def chmod(self, mode: str, path: str, recursive=True):
        if recursive:
            self.exec('chmod -R %s "%s"' % (mode, path))
        else:
            self.exec('chmod %s "%s"' % (mode, path))

    def pm_disable(self, pkg: str, user: int = 0):
        self.exec_on_require(AdbShell.SERVICE_PACKAGE,
                             'pm disable --user %d %s' % (user, pkg))

    def pm_enable(self, pkg: str, user: int = 0):
        self.exec_on_require(AdbShell.SERVICE_PACKAGE,
                             'pm enable --user %d %s' % (user, pkg))

    def pm_clear(self, pkg: str, user: int = 0):
        self.exec_on_require(AdbShell.SERVICE_PACKAGE,
                             'pm clear --user %d %s' % (user, pkg))

    def is_bluetooth_on(self) -> bool:
        return self.settings_get('global', 'bluetooth_on') == '1'

    def set_bluetooth_on(self, on: bool):
        self.exec_on_require(AdbShell.SERVICE_BLUETOOTH,
                             'svc bluetooth %s' % ('enable' if on else 'disable'))

    def is_data_on(self) -> bool:
        return self.settings_get('global', 'mobile_data') == '1'

    def set_data_on(self, on: bool):
        self.exec_on_require(AdbShell.SERVICE_PHONE,
                             'svc data %s' % ('enable' if on else 'disable'))

    # def is_nfc_on(self) -> bool:  # TODO implement later
    #     return False

    def set_nfc_on(self, on: bool):
        self.exec_on_require(AdbShell.SERVICE_NFC,
                             'svc nfc %s' % ('enable' if on else 'disable'))

    def window_policy(self):
        _, lines = self.exec('dumpsys window policy')
        return WindowPolicy(lines)

    def input_keyevent(self, keycode: str):
        self.exec_on_require(AdbShell.SERVICE_INPUT,
                             'input keyevent %s' % keycode)

    def input_swipe(self, x1: int, y1: int, x2: int, y2: int):
        self.exec_on_require(AdbShell.SERVICE_INPUT,
                             'input swipe %d %d %d %d' % (x1, y1, x2, y2))

    def input_tap(self, x: int, y: int):
        self.exec_on_require(AdbShell.SERVICE_INPUT,
                             'input tap %d %d' % (x, y))

    def disable_adb(self):
        self.exec_on_require(AdbShell.SERVICE_SETTINGS,
                             'settings put global adb_enabled 0',
                             allow_offline_on_finish=True)

    def settings_get(self, namespace: str, key: str, user: Union[None, int] = None) -> str:
        if user is None:
            _, lines = self.exec_on_require(AdbShell.SERVICE_SETTINGS,
                                            'settings get %s %s' % (namespace, key))
        else:
            _, lines = self.exec_on_require(AdbShell.SERVICE_SETTINGS,
                                            'settings get --user %d %s %s' % (user, namespace, key))
        if len(lines) > 0:
            return lines[0].strip()
        else:
            return ''

    def settings_put(self, namespace: str, key: str, value: str, user: Union[None, int] = None):
        if user is None:
            self.exec_on_require(AdbShell.SERVICE_SETTINGS,
                                 'settings put %s %s "%s"' % (namespace, key, value))
        else:
            self.exec_on_require(AdbShell.SERVICE_SETTINGS,
                                 'settings put --user %d %s %s "%s"' % (user, namespace, key, value))

    def settings_delete(self, namespace: str, key: str, user: Union[None, int] = None, android_version: int = 11):
        if user is None or android_version < 11:
            self.exec_on_require(AdbShell.SERVICE_SETTINGS,
                                 'settings delete %s %s' % (namespace, key))
        else:
            self.exec_on_require(AdbShell.SERVICE_SETTINGS,
                                 'settings delete --user %d %s %s' % (user, namespace, key))

    def settings_dump_a11(self, namespace: str, user: Union[None, int] = None) -> Dict[str, str]:
        if user is None:
            code, lines = self.exec_on_require(AdbShell.SERVICE_SETTINGS,
                                               'settings list %s' % namespace)
        else:
            code, lines = self.exec_on_require(AdbShell.SERVICE_SETTINGS,
                                               'settings list --user %d %s' % (user, namespace))
        if code != 0:
            return dict()

        pattern = re.compile(r'([^=]+)=(.+)\s*')
        values: Dict[str, str] = dict()
        for line in lines:
            matched = pattern.fullmatch(line)
            if matched:
                key = matched.group(1)
                val = matched.group(2)
                if val != 'null':
                    values[key] = val
        return values

    def settings_dump_a9(self, namespace: str, user: Union[None, int] = None) -> Dict[str, str]:
        values: Dict[str, str] = dict()
        if user is None:
            user = 0
        path = '/data/system/users/%d/settings_%s.xml' % (user, namespace)
        self.wait_for_path_ready(path)
        content = self.cat(path)
        dom_tree = xml.dom.minidom.parseString(content)
        root_element = dom_tree.documentElement
        settings = root_element.getElementsByTagName('setting')
        for setting in settings:
            if setting.hasAttribute('value'):
                values[setting.getAttribute('name')] = setting.getAttribute('value')
        return values

    def settings_restore(self, namespace: str, values: Dict[str, str], user: Union[None, int] = None):
        for key in values:
            self.settings_put(namespace, key, values[key], user=user)

    def cat(self, path: str) -> str:
        code, lines = self.exec('cat "%s"' % path)
        return '\n'.join(lines) if code == 0 else ''

    def is_boot_complete(self) -> bool:
        _, lines = self.exec('getprop dev.bootcomplete')
        return len(lines) > 0 and lines[0] != '0'

    def wait_for_boot_complete(self):
        while not self.is_boot_complete():
            Log.d('wait for boot complete')
            time.sleep(1)

    def am_broadcast(self, action: str):
        self.exec_on_require(AdbShell.SERVICE_ACTIVITY,
                             'am broadcast -a %s' % action,
                             require_boot_complete=True)

    def safe_reboot(self) -> (bool, int):
        while True:
            result, code = self.safe_reboot_no_wait()
            if result:
                if self.wait_for_shell_exit(30):
                    time.sleep(3)  # sleeep another 3 seconds to ensure device rebooted
                    Log.d('device offline after safe reboot cmd sent')
                    break
                else:
                    Log.w('wait for device offline timeout failed')
            else:
                break
        return result, code

    def safe_reboot_no_wait(self) -> (bool, int):
        # cmd = 'am broadcast -a android.intent.action.REBOOT --ei nowait 1 --ei interval 1 --ei window 0'
        code, _ = self.exec_on_require(AdbShell.SERVICE_POWER,
                                       'svc power reboot',
                                       require_boot_complete=True,
                                       allow_offline_on_finish=True)

        return code == 0, code

    def find(self, path: str, min_depth: int = -1, max_depth: int = -1) -> List[str]:
        cmds = ['find %s' % path.replace(' ', '\\ '), ]
        if min_depth >= 0:
            cmds.append('-mindepth %d' % min_depth)
        if max_depth >= 0:
            cmds.append('-maxdepth %d' % min_depth)

        _, lines = self.exec(' '.join(cmds))
        result = [line.strip() for line in lines]
        if len(result) == 1 and AdbShell.PATTERN_FIND_ERROR.match(result[0]):
            return []

        return result

    def pm_list_users(self) -> List[PmListUserInfo]:
        infos = []

        _, lines = self.exec_on_require((AdbShell.SERVICE_PACKAGE, AdbShell.SERVICE_ACTIVITY),
                                        'pm list users')
        for line in lines:
            matched = AdbShell.PATTERN_PM_LIST_USER.match(line)
            if matched:
                infos.append(PmListUserInfo(int(matched.group(1)), matched.group(2), matched.group(3)))

        return infos

    def pm_create_user(self, name: str) -> Union[None, int]:
        _, lines = self.exec_on_require((AdbShell.SERVICE_PACKAGE, AdbShell.SERVICE_ACTIVITY),
                                        'pm create-user "%s"' % name,
                                        require_boot_complete=True,
                                        retry=0)
        if len(lines) < 1:
            Log.e('pm_create_user()# failed. %s' % lines)
            return None

        pattern = re.compile(r'Success: created user id\s+(\d+)')
        matched = pattern.match(lines[0])
        if matched is not None:
            return int(matched.group(1))
        else:
            Log.e('pm_create_user()# failed. %s' % lines)
            return None

    def pm_remove_user(self, user_id: int) -> bool:
        _, lines = self.exec_on_require((AdbShell.SERVICE_PACKAGE, AdbShell.SERVICE_ACTIVITY),
                                        'pm remove-user %d' % user_id,
                                        require_boot_complete=True)
        return len(lines) > 0 and lines[0].startswith('Success:')

    def pm_get_max_users(self) -> int:
        _, lines = self.exec_on_require((AdbShell.SERVICE_PACKAGE, AdbShell.SERVICE_ACTIVITY),
                                        'pm get-max-users')
        if len(lines) <= 0:
            return 0

        matched = re.match(r'Maximum supported users:\s+(\d+)\s*', lines[0])
        return int(matched.group(1)) if matched else 0

    PM_LIST_PACKAGES_ALL = 0
    PM_LIST_PACKAGES_3RD = 1
    PM_LIST_PACKAGES_SYS = 2
    PM_LIST_PACKAGES_ENABLED = 3
    PM_LIST_PACKAGES_DISABLED = 4
    PM_LIST_PACKAGES_FILTER = {
        PM_LIST_PACKAGES_3RD: '-3',
        PM_LIST_PACKAGES_SYS: '-s',
        PM_LIST_PACKAGES_ENABLED: '-e',
        PM_LIST_PACKAGES_DISABLED: '-d'
    }

    def pm_list_packages(self, user: int = 0, flag: int = PM_LIST_PACKAGES_ALL) -> List[PmListPkgInfo]:
        infos: List[PmListPkgInfo] = []

        filter_str = AdbShell.PM_LIST_PACKAGES_FILTER[flag] \
            if flag in AdbShell.PM_LIST_PACKAGES_FILTER \
            else ''

        _, lines = self.exec_on_require(AdbShell.SERVICE_PACKAGE,
                                        'pm list package -f -U --user %d %s' % (user, filter_str))
        for line in lines:
            matched = AdbShell.PATTERN_PM_LIST_PACKAGES.match(line)
            if matched:
                infos.append(PmListPkgInfo(matched.group(2),
                                           int(matched.group(3)),
                                           matched.group(1)))

        return infos

    def pm_install(self, path: str, user: int = 0) -> bool:
        _, lines = self.exec_on_require(AdbShell.SERVICE_PACKAGE,
                                        'pm install -r -t --user %d "%s"' % (user, path))
        if len(lines) == 0:
            return False

        return lines[-1].startswith('Success')

    def pm_install_create(self, user: int = 0) -> Union[None, str]:
        code, lines = self.exec_on_require(AdbShell.SERVICE_PACKAGE,
                                           'pm install-create -r -t --user %d' % user)

        matched = re.fullmatch(r'Success:\s+created\s+install\s+session\s+\[(\d+)]\s*', lines[0])
        return matched.group(1) if code == 0 and matched is not None else None

    def pm_install_write(self, session: str, split_name: Union[None, str] = None,
                         device_path: str = None, host_path: str = None) -> bool:
        if device_path is not None:
            cmd = 'pm install-write %s %s "%s"' % (
                session,
                split_name if split_name is not None else os.path.basename(device_path),
                device_path)
            code, _ = self.exec_on_require(AdbShell.SERVICE_PACKAGE,
                                           cmd)
            return code == 0
        elif host_path is not None:  # TODO has bugs
            def on_input(stdin: Device.ShellIn):
                nonlocal host_path
                with open(host_path, 'rb') as file:
                    while True:
                        buff = file.read(AdbShell.STDIN_BUFF_SIZE)
                        if len(buff) == 0:
                            break
                        stdin.io.write(buff)

            cmd = 'pm install-write -S %d %s %s -' % (
                os.path.getsize(host_path),
                session,
                split_name if split_name is not None else os.path.basename(host_path))
            code, _ = self.exec_on_require(AdbShell.SERVICE_PACKAGE,
                                           cmd,
                                           on_input=on_input)
            return code == 0
        else:
            return False

    def pm_install_remove(self, session: str, split_name: str) -> bool:
        code, _ = self.exec_on_require(AdbShell.SERVICE_PACKAGE,
                                       'pm install-remove %s %s' % (session, split_name))
        return code == 0

    def pm_install_commit(self, session: str) -> bool:
        code, _ = self.exec_on_require(AdbShell.SERVICE_PACKAGE,
                                       'pm install-commit %s' % session,
                                       retry=0)
        return code == 0

    def pm_install_abandon(self, session: str) -> bool:
        code, _ = self.exec_on_require(AdbShell.SERVICE_PACKAGE,
                                       'pm install-abandon %s' % session,
                                       retry=0)
        return code == 0

    def pm_install_session(self, user: int = 0):
        return AdbShell.InstallSession(self, user)

    def pm_install_existing(self, user: int, pkg: str):
        self.exec_on_require(AdbShell.SERVICE_PACKAGE,
                             'pm install-existing --user %d %s' % (user, pkg))

    def am_is_user_stopped(self, user: int) -> bool:
        _, lines = self.exec_on_require([AdbShell.SERVICE_ACTIVITY, AdbShell.SERVICE_PACKAGE],
                                        'am is-user-stopped %d' % user)
        if len(lines) == 0:
            return False

        return lines[0].startswith('true')

    def am_get_current_user(self) -> Union[None, int]:
        _, lines = self.exec_on_require([AdbShell.SERVICE_ACTIVITY, AdbShell.SERVICE_PACKAGE],
                                        'am get-current-user')
        if len(lines) == 0:
            return None

        str_code = lines[0].strip()
        return int(str_code) if str_code.isnumeric() else None

    def am_start_user(self, user: int) -> bool:
        _, lines = self.exec_on_require([AdbShell.SERVICE_ACTIVITY, AdbShell.SERVICE_PACKAGE],
                                        'am start-user %d' % user)
        if len(lines) == 0:
            return False

        return lines[-1].startswith('Success')

    def am_stop_user(self, user: int):
        self.exec_on_require([AdbShell.SERVICE_ACTIVITY, AdbShell.SERVICE_PACKAGE],
                             'am stop-user %d' % user)

    def am_switch_user(self, user: int):
        self.exec_on_require([AdbShell.SERVICE_ACTIVITY, AdbShell.SERVICE_PACKAGE],
                             'am switch-user %d' % user)

    def service_check(self, service: str) -> bool:
        code, lines = self.exec('service check %s' % service)
        return code == 0 and re.fullmatch(r'\s*Service\s+%s:\s+found\s*' % service, lines[0]) is not None

    def locksettings_get_disabled(self, user: int) -> bool:
        _, lines = self.exec_on_require(AdbShell.SERVICE_LOCK_SETTINGS,
                                        'locksettings get-disabled --user %d' % user)
        if len(lines) == 0:
            return False

        return re.fullmatch(r'Error while executing command: get-disabled\s*', lines[0]) is None

    def wm_get_density(self) -> int:
        code, lines = self.exec_on_require(AdbShell.SERVICE_WINDOW,
                                           'wm density')
        if code != 0:
            return 320  # default

        density: Union[None, int] = None
        density_type: Union[None, str] = None
        for line in lines:
            for pattern in AdbShell.PATTERN_WM_DENSITY:
                matched = pattern.fullmatch(line)
                if matched:
                    if density_type is None or density_type == 'Physical':
                        density_type = matched.group(1)
                        density = int(matched.group(2))
        return density if density is not None else 320

    def wm_set_density(self, density: int):
        self.exec_on_require(AdbShell.SERVICE_WINDOW,
                             'wm density %d' % density)

    def uimode_get_night_mode(self) -> int:
        _, lines = self.exec_on_require(AdbShell.SERVICE_UIMODE,
                                        'service call uimode 6')
        if len(lines) <= 0:
            return AdbShell.NIGHT_MODE_OFF

        matched = re.fullmatch(r'Result:\s+Parcel\(00000000\s+(\d+)\s+.*\)\s*', lines[0])
        if matched:
            return int(matched.group(1))
        else:
            return AdbShell.NIGHT_MODE_OFF

    def uimode_set_night_mode(self, mode: int):
        self.exec_on_require(AdbShell.SERVICE_UIMODE,
                             'service call uimode 5 i32 %d' % mode)

    def is_night_display_on(self, user: Union[None, int] = None) -> bool:
        on = self.settings_get('secure', 'night_display_activated', user=user)
        return on == '1'

    def set_night_display_on(self, on: bool, user: Union[None, int] = None):
        self.settings_put('secure', 'night_display_activated', '1' if on else '0', user=user)

    def get_night_display_color_temperature(self, user: Union[None, int] = None) -> int:
        val = self.settings_get('secure', 'night_display_color_temperature', user=user)
        return int(val) if val.isnumeric() else 3000

    def set_night_display_color_temperature(self, temperature: int, user: Union[None, int] = None):
        self.settings_put('secure', 'night_display_color_temperature', str(temperature), user=user)

    def get_screen_brightness_mode(self, user: Union[None, int] = None) -> int:
        val = self.settings_get('system', 'screen_brightness_mode', user=user)
        return int(val) if val.isnumeric() else 0

    def set_screen_brightness_mode(self, brightness: int, user: Union[None, int] = None):
        self.settings_put('system', 'screen_brightness_mode', str(brightness), user=user)

    def get_screen_brightness(self, user: Union[None, int] = None) -> int:
        val = self.settings_get('system', 'screen_brightness', user=user)
        return int(val) if val.isnumeric() else 0

    def set_screen_brightness(self, brightness: int, user: Union[None, int] = None):
        self.settings_put('system', 'screen_brightness', str(brightness), user=user)

    class InstallSession(object):
        def __init__(self, shell, user: int = 0):
            self.shell: AdbShell = shell
            self.user = user
            self.session: Union[None, str] = None
            self.result = True
            self.commit = True

        def __enter__(self):
            self.create_session()
            return self

        def __exit__(self, exc_type, exc_val, exc_tb):
            self.commit_session()

        def create_session(self) -> bool:
            self.session = self.shell.pm_install_create(self.user)
            self.result = self.session is not None
            return self.result

        def commit_session(self) -> bool:
            if self.session is None:
                return False

            if self.result and self.commit:
                self.result = self.shell.pm_install_commit(self.session)
            else:
                self.shell.pm_install_abandon(self.session)
            return self.result

        def write(self, name: Union[None, str] = None,
                  device_path: str = None, host_path: str = None) -> bool:
            if self.result:
                self.result = self.shell.pm_install_write(self.session, name, device_path, host_path)
            return self.result

        def abandon(self):
            self.commit = False
